{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyalink.alink import *\n",
    "useLocalEnv(1)\n",
    "\n",
    "from utils import *\n",
    "import os\n",
    "import pandas as pd\n",
    "\n",
    "DATA_DIR = ROOT_DIR + \"movielens\" + os.sep + \"ml-100k\" + os.sep\n",
    "\n",
    "RATING_FILE = \"u.data\";\n",
    "ITEM_FILE = \"u.item\";\n",
    "\n",
    "RATING_SCHEMA_STRING = \"user_id long, item_id long, rating float, ts long\";\n",
    "\n",
    "ITEM_SCHEMA_STRING = \"item_id long, title string, \"\\\n",
    "    + \"release_date string, video_release_date string, imdb_url string, \"\\\n",
    "    + \"unknown int, action int, adventure int, animation int, \"\\\n",
    "    + \"children int, comedy int, crime int, documentary int, drama int, \"\\\n",
    "    + \"fantasy int, film_noir int, horror int, musical int, mystery int, \"\\\n",
    "    + \"romance int, sci_fi int, thriller int, war int, western int\";\n",
    "\n",
    "\n",
    "def getSourceRatings() :\n",
    "    return TsvSourceBatchOp()\\\n",
    "            .setFilePath(DATA_DIR + RATING_FILE)\\\n",
    "            .setSchemaStr(RATING_SCHEMA_STRING);\n",
    "\n",
    "\n",
    "def getStreamSourceRatings() :\n",
    "    return TsvSourceStreamOp()\\\n",
    "            .setFilePath(DATA_DIR + RATING_FILE)\\\n",
    "            .setSchemaStr(RATING_SCHEMA_STRING);\n",
    "\n",
    "def getSourceItems() :\n",
    "    return CsvSourceBatchOp()\\\n",
    "            .setFieldDelimiter(\"|\")\\\n",
    "            .setFilePath(DATA_DIR + ITEM_FILE)\\\n",
    "            .setSchemaStr(ITEM_SCHEMA_STRING);\n",
    "\n",
    "\n",
    "def getStreamSourceItems() :\n",
    "    return CsvSourceStreamOp()\\\n",
    "            .setFieldDelimiter(\"|\")\\\n",
    "            .setFilePath(DATA_DIR + ITEM_FILE)\\\n",
    "            .setSchemaStr(ITEM_SCHEMA_STRING);\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import datetime\n",
    "\n",
    "@udf(input_types=[DataTypes.BIGINT()], result_type=DataTypes.TIMESTAMP(3))\n",
    "def from_unix_timestamp(ts):\n",
    "    return datetime.datetime.fromtimestamp(ts)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_1_3\n",
    "\n",
    "ratings = getSourceRatings();\n",
    "\n",
    "ratings.firstN(5).print();\n",
    "\n",
    "ratings\\\n",
    "    .link(\n",
    "        UDFBatchOp()\\\n",
    "            .setFunc(from_unix_timestamp)\\\n",
    "            .setSelectedCols([\"ts\"])\\\n",
    "            .setOutputCol(\"ts\")\n",
    "    )\\\n",
    "    .firstN(5)\\\n",
    "    .print();\n",
    "\n",
    "BatchOperator.registerFunction(\"from_unix_timestamp\", from_unix_timestamp);\n",
    "\n",
    "ratings\\\n",
    "    .select(\"user_id, item_id, rating, from_unix_timestamp(ts) AS ts\")\\\n",
    "    .firstN(5)\\\n",
    "    .print();\n",
    "\n",
    "ratings.registerTableName(\"ratings\");\n",
    "\n",
    "BatchOperator\\\n",
    "    .sqlQuery(\"SELECT user_id, item_id, rating, from_unix_timestamp(ts) AS ts FROM ratings\")\\\n",
    "    .firstN(5)\\\n",
    "    .print();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_1_4\n",
    "\n",
    "ratings = getStreamSourceRatings();\n",
    "\n",
    "ratings = ratings.filter(\"user_id=1 AND item_id<5\");\n",
    "\n",
    "ratings.print();\n",
    "\n",
    "StreamOperator.execute();\n",
    "\n",
    "ratings\\\n",
    "    .link(\n",
    "        UDFStreamOp()\\\n",
    "            .setFunc(from_unix_timestamp)\\\n",
    "            .setSelectedCols([\"ts\"])\\\n",
    "            .setOutputCol(\"ts\")\n",
    "    )\\\n",
    "    .print();\n",
    "\n",
    "StreamOperator.execute();\n",
    "\n",
    "StreamOperator.registerFunction(\"from_unix_timestamp\", from_unix_timestamp);\n",
    "\n",
    "ratings\\\n",
    "    .select(\"user_id, item_id, rating, from_unix_timestamp(ts) AS ts\")\\\n",
    "    .print();\n",
    "\n",
    "StreamOperator.execute();\n",
    "\n",
    "ratings.registerTableName(\"ratings\");\n",
    "\n",
    "StreamOperator\\\n",
    "    .sqlQuery(\"SELECT user_id, item_id, rating, from_unix_timestamp(ts) AS ts FROM ratings\")\\\n",
    "    .print();\n",
    "\n",
    "StreamOperator.execute();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@udtf(input_types=[DataTypes.STRING()], result_types=[DataTypes.STRING(), DataTypes.INT()])\n",
    "def doc_word_count(s):\n",
    "    dict = {}\n",
    "    for t in s.split() :\n",
    "        if t in dict :\n",
    "            dict[t] = dict[t] + 1\n",
    "        else :\n",
    "            dict[t] = 1\n",
    "    \n",
    "    for k in dict :\n",
    "        yield k, dict[k]\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2_3\n",
    "\n",
    "items = getSourceItems();\n",
    "\n",
    "items.select(\"item_id, title\").lazyPrint(10, \"<- original data ->\");\n",
    "\n",
    "words = items\\\n",
    "    .link(\n",
    "        UDTFBatchOp()\\\n",
    "            .setFunc(doc_word_count)\\\n",
    "            .setSelectedCols([\"title\"])\\\n",
    "            .setOutputCols([\"word\", \"cnt\"])\\\n",
    "            .setReservedCols([\"item_id\"])\n",
    "    );\n",
    "\n",
    "words.lazyPrint(20, \"<- after word count ->\");\n",
    "\n",
    "words.groupBy(\"word\", \"word, SUM(cnt) AS cnt\")\\\n",
    "    .orderBy(\"cnt\", 20, order='desc')\\\n",
    "    .print();\n",
    "\n",
    "BatchOperator.registerFunction(\"doc_word_count\", doc_word_count);\n",
    "\n",
    "items.registerTableName(\"items\");\n",
    "\n",
    "BatchOperator\\\n",
    "    .sqlQuery(\"SELECT item_id, word, cnt FROM items, \"\n",
    "              + \"LATERAL TABLE(doc_word_count(title)) as T(word, cnt)\")\\\n",
    "    .firstN(20)\\\n",
    "    .print();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2_4\n",
    "\n",
    "items = getStreamSourceItems();\n",
    "\n",
    "items = items.select(\"item_id, title\").filter(\"item_id<4\");\n",
    "\n",
    "items.print();\n",
    "StreamOperator.execute();\n",
    "\n",
    "words = items\\\n",
    "    .link(\n",
    "        UDTFStreamOp()\\\n",
    "            .setFunc(doc_word_count)\\\n",
    "            .setSelectedCols([\"title\"])\\\n",
    "            .setOutputCols([\"word\", \"cnt\"])\\\n",
    "            .setReservedCols([\"item_id\"])\n",
    "    );\n",
    "\n",
    "words.print();\n",
    "StreamOperator.execute();\n",
    "\n",
    "StreamOperator.registerFunction(\"doc_word_count\", doc_word_count);\n",
    "\n",
    "items.registerTableName(\"items\");\n",
    "\n",
    "StreamOperator\\\n",
    "    .sqlQuery(\"SELECT item_id, word, cnt FROM items, \"\n",
    "              + \"LATERAL TABLE(doc_word_count(title)) as T(word, cnt)\")\\\n",
    "    .print();\n",
    "\n",
    "StreamOperator.execute();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
